"""
Case Type   : ustore_DQL_SET_OPERATIONS
Case Name   : 场景测试—union在OALP场景测试
Create At   : 2022-05-12
Owner       : nanyang12
Description :
    1.创建表
    2.使用copy命令插入数据
    3.union在OALP场景测试
    4.环境清理
Expect      :
    1.创建成功
    2.数据插入成功
    3.成功
    4.清理成功
History     :
"""

import os
import subprocess
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Ustore01(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----- {os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_root = Node('PrimaryRoot')
        self.com = Common()
        self.constant = Constant()
        self.table_name1 = 't_ustore_olap_0001_01'
        self.table_name2 = 't_ustore_olap_0001_02'
        self.copy_file1 = 'customer_33.txt'
        self.copy_file2 = 'web_site_33.txt'
        self.cur_path_1 = os.path.join(macro.SCRIPTS_PATH, self.copy_file1)
        self.cur_path_2 = os.path.join(macro.SCRIPTS_PATH, self.copy_file2)
        self.parent_path = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.target_path1 = os.path.join(self.parent_path, self.copy_file1)
        self.target_path2 = os.path.join(self.parent_path, self.copy_file2)

    def test_join(self):
        step_txt = '-----step1:创建表;expect: 创建成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""drop table if exists {self.table_name1}  CASCADE ;
                    drop table if exists {self.table_name2}  CASCADE ;
        create table {self.table_name1}
        (
            c_customer_sk             integer               not null,
            c_customer_id             char(16)              not null,
            c_current_cdemo_sk        integer                       ,
            c_current_hdemo_sk        integer                       ,
            c_current_addr_sk         integer                       ,
            c_first_shipto_date_sk    integer                       ,
            c_first_sales_date_sk     integer                       ,
            c_salutation              char(10)                      ,
            c_first_name              char(20)                      ,
            c_last_name               char(30)                      ,
            c_preferred_cust_flag     char(1)                       ,
            c_birth_day               integer                       ,
            c_birth_month             integer                       ,
            c_birth_year              integer                       ,
            c_birth_country           varchar(20)                   ,
            c_login                   char(13)                      ,
            c_email_address           char(50)                      ,
            c_last_review_date        char(10)
        )  WITH (STORAGE_TYPE=USTORE);
        create table {self.table_name2}
        (
            web_site_sk               integer               not null,
            web_site_id               char(16)              not null,
            web_rec_start_date        date                          ,
            web_rec_end_date          date                          ,
            web_name                  varchar(50)                   ,
            web_open_date_sk          integer                       ,
            web_close_date_sk         integer                       ,
            web_class                 varchar(50)                   ,
            web_manager               varchar(40)                   ,
            web_mkt_id                integer                       ,
            web_mkt_class             varchar(50)                   ,
            web_mkt_desc              varchar(100)                  ,
            web_market_manager        varchar(100)                   ,
            web_company_id            integer                  ,
            web_company_name          char(50)                      ,
            web_street_number         char(10)                      ,
            web_street_name           varchar(60)                   ,
            web_street_type           char(15)                      ,
            web_suite_number          char(50)                      ,
            web_city                  varchar(60)                   ,
            web_county                varchar(30)                   ,
            web_state                 char(20)                       ,
            web_zip                   char(50)                      ,
            web_country               varchar(20)                   ,
            web_gmt_offset            decimal(5,2)                  ,
            web_tax_percentage        decimal(5,2)                  
         )
         WITH (STORAGE_TYPE=USTORE) partition by range(web_rec_start_date)
        (
         partition p1 values less than('1990-01-01'),
         partition p2 values less than('1995-01-01'),
         partition p3 values less than('2000-01-01'),
         partition p4 values less than('2005-01-01'),
         partition p5 values less than('2010-01-01'),
         partition p6 values less than('2015-01-01'),
         partition p7 values less than(maxvalue)
        );"""
        self.log.info(cmd_sql)
        create_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(create_result)
        self.assertTrue(
            create_result.count(self.constant.TABLE_CREATE_SUCCESS) == 2,
            '执行失败:' + step_txt)

        step_txt = '-----step2:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(step_txt)
        text = '-----step2.1: 获取当前主机的IP地址;expect: 成功-----'
        self.log.info(text)
        cmd = "ifconfig -a|grep inet6 -a2|grep broadcast|awk '{print $2}'"
        self.log.info(cmd)
        self.cur_ip = subprocess.getoutput(cmd).strip()
        self.log.info(self.cur_ip)

        text = '-----step2.2: 复制文件到数据库主机对应目录 expect: 成功-----'
        self.log.info(text)
        if self.cur_ip == self.pri_sh.node.ssh_host:
            self.log.info('当前主机是数据库主机，无需远程复制')
            cmd = f"cp {self.cur_path_1} {self.parent_path};" \
                f"cp {self.cur_path_2} {self.parent_path};"
            self.log.info(cmd)
            res = self.pri_sh.node.sh(cmd).result()
            self.log.info(res)
        else:
            self.com.scp_file(self.pri_root, self.copy_file1,
                              self.parent_path)
            self.com.scp_file(self.pri_root, self.copy_file2,
                              self.parent_path)
            chmod_cmd = f"chmod 777 -R {self.parent_path};" \
                        f"ls -al {self.parent_path}"
            self.log.info(chmod_cmd)
            chmod_result = self.pri_root.sh(chmod_cmd).result()
            self.log.info(chmod_result)
        text = '-----step2.3: 查看数据库主机目标路径下文件 expect: 成功-----'
        self.log.info(text)
        cmd = f"ls -l {self.parent_path}| grep -E '{self.copy_file1}';" \
            f"ls -l {self.parent_path}| grep -E '{self.copy_file2}';"
        self.log.info(cmd)
        res = self.pri_root.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('No such file or directory', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file1}', res, '文件拷贝失败')
        self.assertIn(f'{self.copy_file2}', res, '文件拷贝失败')

        text = '-----step2.4:使用copy命令插入数据，expect: 数据插入成功-----'
        self.log.info(text)
        self.log.info('---------------使用copy命令插入数据------------------')
        copy_sql = f"copy {self.table_name1} from '{self.target_path1}' " \
            f"delimiters ',';" \
            f"copy {self.table_name2} from '{self.target_path2}' " \
            f"delimiters ',';"
        self.log.info(copy_sql)
        copy_result = self.pri_sh.execut_db_sql(copy_sql)
        self.log.info(copy_result)
        self.assertIn('COPY 30', copy_result, '执行失败:' + step_txt)
        self.assertIn('COPY 10000', copy_result, '执行失败:' + step_txt)

        step_txt = '-----step3: union在OALP场景测试 expect: 成功-----'
        self.log.info(step_txt)
        cmd_sql = f"""
        SELECT * FROM (
        select cast((c_birth_year  || '-' || c_birth_month
                      || '-' || c_birth_day ) as date) a,
               case
                 when c_salutation is null then
                  coalesce(c_preferred_cust_flag, '', trim(c_last_name))
                 when c_salutation like 'Dr%' then
                  c_salutation || 'A'
                 when c_salutation like 'Miss%' then
                  c_salutation || trim(c_first_name)
                 when c_salutation like 'Mr.%' then
                  c_salutation || trim(c_last_name)
                 when c_salutation like 'Mrs.%' then
                  c_salutation || substr(trim(c_first_name), 2, 3)
                 when c_salutation like 'Ms.%' then
                  c_salutation || substr(trim(c_last_name), 2, 3)
                 else
                  c_salutation || 'B'
               end b,
               coalesce(substr(c_email_address, 3, 4),
                        max(trim(c_login)),
                        min(distinct trim(c_last_review_date))) g,
               (c_preferred_cust_flag ) d,
               c_current_hdemo_sk / c_birth_day e,
               1 c,
               3 f
          from {self.table_name1}
         where c_birth_day = c_birth_month
           and c_birth_day is not null
           and c_birth_year is not null
           and c_birth_month is not null
         group by grouping sets(cube(case
                 when c_salutation is null then
                  coalesce(c_preferred_cust_flag, '', trim(c_last_name))
                 when c_salutation like 'Dr%' then
                  c_salutation || 'A'
                 when c_salutation like 'Miss%' then
                  c_salutation || trim(c_first_name)
                 when c_salutation like 'Mr.%' then
                  c_salutation || trim(c_last_name)
                 when c_salutation like 'Mrs.%' then
                  c_salutation || substr(trim(c_first_name), 2, 3)
                 when c_salutation like 'Ms.%' then
                  c_salutation || substr(trim(c_last_name), 2, 3)
                 else
                  c_salutation || 'B'
               end, cast((c_birth_year  || '-' || c_birth_month
                      || '-' || c_birth_day ) as date), c_current_hdemo_sk / c_birth_day), rollup(cast((c_birth_year  || '-' || c_birth_month
                      || '-' || c_birth_day ) as date), case
                 when c_salutation is null then
                  coalesce(c_preferred_cust_flag, '', trim(c_last_name))
                 when c_salutation like 'Dr%' then
                  c_salutation || 'A'
                 when c_salutation like 'Miss%' then
                  c_salutation || trim(c_first_name)
                 when c_salutation like 'Mr.%' then
                  c_salutation || trim(c_last_name)
                 when c_salutation like 'Mrs.%' then
                  c_salutation || substr(trim(c_first_name), 2, 3)
                 when c_salutation like 'Ms.%' then
                  c_salutation || substr(trim(c_last_name), 2, 3)
                 else
                  c_salutation || 'B'
               end, 1), c_email_address),
                  substr(c_email_address, 3, 4),(c_preferred_cust_flag )
        HAVING c_current_hdemo_sk / c_birth_day is not null AND cast((c_birth_year  || '-' || c_birth_month
                      || '-' || c_birth_day ) as date) IS NOT NULL
        union
        select web_rec_start_date,
               web_name,
               length(web_mkt_desc)::varchar ,
               web_mkt_id::varchar,
               max(distinct web_gmt_offset),
               a,
               web_tax_percentage
          from (select *, '' ::int as a from {self.table_name2})
         where web_company_id <> a and web_mkt_id > web_site_sk
         group by web_rec_start_date,
               web_name,
               length(web_mkt_desc) ,
               web_mkt_id ,
               a,
               web_tax_percentage)
         group by a,b,c,d,e,f,g
         HAVING c =1
         order by 1,2,3,4,5,6,7 limit 20;"""
        self.log.info(cmd_sql)
        query_result = self.pri_sh.execut_db_sql(cmd_sql)
        self.log.info(query_result)
        self.assertIn('20 rows', query_result, '执行失败:' + step_txt)

    def tearDown(self):
        step_txt = '-----step4: 环境清理 expect:清理成功-----'
        self.log.info(step_txt)
        self.log.info('删除表')
        cmd_sql = self.pri_sh.execut_db_sql(f"""
                    drop table {self.table_name1} cascade;
                    drop table {self.table_name2} cascade;""")
        self.log.info(cmd_sql)
        text = '-----删除文件-----'
        self.log.info(text)
        rm_cmd = f"rm -rf {self.target_path1};" \
            f"rm -rf {self.target_path2};"
        self.log.info(rm_cmd)
        self.com.get_sh_result(self.pri_root, rm_cmd)
        for s in [self.target_path1, self.target_path2]:
            cmd = f'if [ -f {s} ]; then echo "does exists"; ' \
                f'else echo "not exists"; fi'
            file_judgments = self.com.get_sh_result(self.pri_root, cmd)
            self.assertIn('not exists', file_judgments, "清理失败" + text)
        self.assertTrue(
            cmd_sql.count(self.constant.DROP_TABLE_SUCCESS) == 2,
            '执行失败:' + step_txt)
        self.log.info(f'----- {os.path.basename(__file__)} end-----')
